#include <assert.h>
#include <stdlib.h>

#include <common/macro/build.h>
#include <common/cache.h>
#include <common/common.h>
#include <common/util.h>
#include <common/macro/array.h>
#include <common/macro/debug.h>

#include <database/database.h>

#include <maintenance/vacuum.h>

#include <transaction/abortlog.h>
#include <transaction/transaction.h>
#include <transaction/transactionlist.h>


API_FILELOCAL inline void cache_removal_callback(void *key, void *data, void *ctx) {
	debugmsg("Cache callback (%p, %p, %p)\n", key, data, ctx);
	free(key);
	free(data);
}
API_FILELOCAL inline int cache_cmp_transaction_id(transaction_id_t *t1, transaction_id_t *t2) {
	assert(t1 != NULL);
	assert(t2 != NULL);

	debugmsg("Cache key comparison (%p -> %lu, %p -> %lu)\n", t1, *t1, t2, *t2);

	if (*t1 > *t2)
		return 1;
	if (*t1 < *t2)
		return -1;
	return 0;
}

API_FILELOCAL transaction_t *transaction_new(database_t *database, transaction_isolation_e isolation) {
	transaction_t *transaction;

	assert(database != NULL);

	debugmsg("New transaction (%p, isolation %d)\n", database, isolation);

	transaction = malloc(sizeof(*transaction));
	transaction->database = database;
	transaction->isolation = isolation;

	/* TODO, do not hardcode cache size. */
	transaction->transaction_list = transaction_list_new();
	transaction->abort_log_cache = cache_new(1024, (generic_cmp_f)cache_cmp_transaction_id);
	cache_set_callback(transaction->abort_log_cache, cache_removal_callback, NULL);

	transaction->phase = UNKNOWN;
	transaction->command_id = 0;

	return transaction;
}

API_LIBLOCAL int transaction_to_active(transaction_t *transaction) {
	assert(transaction != NULL);
	if (transaction->phase == ACTIVE) { return 0; } 
	assert(transaction->phase == UNKNOWN);

	debugmsg("Setting status to ACTIVE (%p, #%lu)\n", transaction, transaction->id);

	transaction->phase = ACTIVE;
	return transaction_log_write_phase(transaction->database->transaction_log, transaction->id, transaction->phase);
}

API_LIBLOCAL int transaction_to_partially_commited(transaction_t *transaction) {
	assert(transaction != NULL);
	if (transaction->phase == PARTIALLY_COMMITED) { return 0; } 
	assert(transaction->phase == ACTIVE);

	debugmsg("Setting status to PARTIALLY_COMMITED (%p, #%lu)\n", transaction, transaction->id);

	transaction->phase = PARTIALLY_COMMITED;
	return transaction_log_write_phase(transaction->database->transaction_log, transaction->id, transaction->phase);
}

API_LIBLOCAL int transaction_to_commited(transaction_t *transaction) {
	assert(transaction != NULL);
	if (transaction->phase == COMMITED) { return 0; } 
	assert(transaction->phase == PARTIALLY_COMMITED);

	debugmsg("Setting status to COMMITED (%p, #%lu)\n", transaction, transaction->id);

	transaction->phase = COMMITED;
	return transaction_log_write_phase(transaction->database->transaction_log, transaction->id, transaction->phase);
}

API_LIBLOCAL int transaction_to_failed(transaction_t *transaction) {
	assert(transaction != NULL);
	if (transaction->phase == FAILED) { return 0; } 
	assert(transaction->phase == ACTIVE || transaction->phase == PARTIALLY_COMMITED);

	debugmsg("Setting status to FAILED (%p, #%lu)\n", transaction, transaction->id);

	transaction->phase = FAILED;
	return transaction_log_write_phase(transaction->database->transaction_log, transaction->id, transaction->phase);
}

API_LIBLOCAL int transaction_to_aborted(transaction_t *transaction) {
	assert(transaction != NULL);
	if (transaction->phase == ABORTED) { return 0; } 
	assert(transaction->phase == FAILED);

	debugmsg("Setting status to ABORTED (%p, #%lu)\n", transaction, transaction->id);

	transaction->phase = ABORTED;
	return transaction_log_write_phase(transaction->database->transaction_log, transaction->id, transaction->phase);
}

API_LIBLOCAL int transaction_force_abort(transaction_t *transaction) {
	assert(transaction != NULL);

	/*
	 * This should only be used as a failover, when normal transaction processing fails.
	 * That means most likely an internal error, data corruption or a deadlock forcing
	 * a query to fail.
	 */
	debugmsg("Forcibly setting transaction as ABORTED (%p, #%lu)\n", transaction, transaction->id);

	transaction->phase = ABORTED;
	return transaction_log_write_phase(transaction->database->transaction_log, transaction->id, transaction->phase);
}

API_LIBLOCAL void transaction_destroy(transaction_t *transaction) {
	if (transaction == NULL) return;

	debugmsg("Destroying transaction (%p, #%lu)\n", transaction, transaction->id);

	transaction_list_destroy(transaction->transaction_list);
	cache_destroy(transaction->abort_log_cache);
	free(transaction);
}

API_LIBLOCAL void transaction_refresh(transaction_t *transaction) {
	assert(transaction != NULL);
	
	debugmsg("Refreshing transaction (%p, #%lu)\n", transaction, transaction->id);

	switch (transaction->isolation) {
		case READ_COMMITTED:
			debugmsg("Isolation is READ COMMITED, so we continue.\n");
			break;

		/* These are not implemented.*/
		case REPEATABLE_READ:
		case DIRTY_READ:
		case SERIALIZED:
			debugmsg("Isolation is not implemented.\n");
			return;
	}

	debugmsg("Performing refresh.\n");

	/* Refresh the transaction list. */
	transaction_list_refresh(transaction->transaction_list, transaction->database->transaction_log);

	/* Ditch abort log cache. */
	cache_flush(transaction->abort_log_cache);
}

API_LIBLOCAL int transaction_visibility(transaction_t *transaction, transaction_id_t transaction_id, transaction_phase_e *phase) {
	transaction_id_t *cached_id;
	transaction_phase_e *cached_phase;

	assert(transaction != NULL);
	assert(phase != NULL);

	debugmsg("This transaction (%p, #%d) verifies visibility of #%d\n", transaction, transaction->id, transaction_id);
	/*
	 * 1. Verify sequence.
	 * 
	 * The transaction we are verifying must not have been created after this
	 * list was taken. If it is, it is definitely future.
	 */
	if (transaction_list_compare_ids(transaction->transaction_list, transaction->transaction_list->last_id, transaction_id) <= 0) {
		*phase = ACTIVE;
		debugmsg("The queried transaction is definitely newer.\n");
		return 0; 
	}

	/*
	 * 2. Query the transaction list.
	 *
	 * This will verify whether a transaction is in the current transaction list,
	 * or possibly known-uncommited because it is far in the future.
	 */
	if (transaction_list_query(transaction->transaction_list, transaction_id, phase)) {	
		debugmsg("The currently held list returns that it is %d.\n", *phase);
		return 0;
	}
	
	/*
	 * 3. Check abort log.
	 *
	 * We implicitely assume that a transaction is commited, unless proven otherwise.
	 * The abort log is the method to verify this. Since abort log should not be
	 * consulted too often (we assume few aborts on lots of commits), but might grow
	 * large, we do not read it whole. We also cache the status of queried
	 * transactions.
	 */
	if (cache_lookup(transaction->abort_log_cache, &transaction_id, (void**)(&cached_phase)) == 0) {
		*phase = *cached_phase;
		debugmsg("The abort log cache returns that it is %d.\n", *phase);
		return 0;
	}

	/* TODO, caching. */
	if (abort_log_check(transaction->database->abort_log, transaction_id)) {
		debugmsg("The transaction is known ABORTED.\n");
		*phase = ABORTED;
	} else {
		*phase = COMMITED;
		debugmsg("The transaction is assumed COMMITED (%d).\n", *phase);
	}
	cached_id = malloc(sizeof(*cached_id));
	*cached_id = transaction_id;
	cached_phase = malloc(sizeof(*cached_phase));
	*cached_phase = *phase;

	cache_insert(transaction->abort_log_cache, cached_id, cached_phase);

	return 0;
}

API_LIBLOCAL int transaction_deleteability(transaction_t *transaction, transaction_id_t transaction_id, transaction_phase_e *phase) {
	transaction_visibility(transaction, transaction_id, phase);
	if (*phase == ACTIVE) {
		/* If it is active, let's wait for it. */
		debugmsg("Waiting for: %u\n", transaction_id);

		/* TODO, cache results of waits. Especially since we get ACTIVE even if the transaction is commited from deletabilitys point of view. */
		if (transaction_log_lock_item(transaction->database->transaction_log, transaction_id) != 0) {
			/* TODO, verify deadlock! */
			return -1;
		}

		/* Read the value, pass to *phase */
		transaction_log_read_phase(transaction->database->transaction_log, transaction_id, phase);

		transaction_log_unlock_item(transaction->database->transaction_log, transaction_id);
	}

	return 0;
}

API_PUBLIC transaction_t *transaction_begin(database_t *database, transaction_isolation_e isolation) {
	transaction_phase_e phase;
	transaction_id_t old_id, vacuum_id;
	transaction_t *transaction;

	transaction = transaction_new(database, isolation);
	if (transaction == NULL)
		return transaction;

	/* LOCK{ transaction_log last id } */
	debugmsg("Locking current.\n");
	transaction_log_writelock_current(transaction->database->transaction_log);

	transaction_log_get_current(transaction->database->transaction_log, &transaction->id);
	debugmsg("Current transaction ID: %d\n", transaction->id);

	/* LOCK{ transaction_log transaction } */
	transaction_log_lock_item(transaction->database->transaction_log, transaction->id);
	
	/* We have now acquired lock on the item. Let's see what was there before us. */
	old_id = transaction_log_replaces(transaction->database->transaction_log, transaction->id);

	transaction_log_read_phase(transaction->database->transaction_log, old_id, &phase);
	debugmsg("Phase of %d: %d\n", old_id, phase);
	if (phase != COMMITED) {
		debugmsg("Not commited (%d) -- adding to abort log.\n", old_id);
		/*
		 * It's not a commited transaction, and we managed to lock it. That means
		 * it's either aborted or crashed transaction, and should move to abortlog.
		 */
		abort_log_add(transaction->database->abort_log, old_id);
		abort_log_flush(transaction->database->abort_log);
	}

	/* Only now, after the old transaction is safely dealt with, can we update the current ID. */
	transaction_log_set_current(transaction->database->transaction_log, transaction->id + 1);
	debugmsg("Current transaction ID set to: %d\n", transaction->id + 1);
	
	transaction_to_active(transaction);
	transaction_log_flush(transaction->database->transaction_log);

	transaction_log_readlock_vacuum(transaction->database->transaction_log);
	transaction_log_get_vacuum(transaction->database->transaction_log, &vacuum_id);
	transaction_log_unlock_vacuum(transaction->database->transaction_log);

	transaction_log_unlock_current(transaction->database->transaction_log);
	debugmsg("Current unlocked.\n");
	/* UNLOCK{ transaction_log last id } */

	if (transaction->id + 1 == vacuum_id) {
		debugmsg("Enforcing vacuum.\n");
		maintenance_vacuum_database(transaction);
	}

	return transaction;
}

API_PUBLIC int transaction_commit(transaction_t *transaction) {
	transaction_phase_e phase;

	/* Verify if the transaction wasn't condemned. */
	if (transaction->phase != ACTIVE) {
		/* Forcibly rollback instead of commit. */
		transaction_rollback(transaction);
		return -1;
	}

	/* Verify that we can commit. That means that our tx was not ditched as aborted. */
	transaction_log_read_phase(transaction->database->transaction_log, transaction->id, &phase);
	if (transaction->phase != phase) {
		/*
		 * What's in the transaction log does not correlate to what we think.
		 * This is a grave failure, bail out as aborted immediatelly.
		 */
		transaction_force_abort(transaction);
		transaction_log_flush(transaction->database->transaction_log);
		transaction_log_unlock_item(transaction->database->transaction_log, transaction->id);
		transaction_destroy(transaction);
		return -1;
	}

	transaction_to_partially_commited(transaction);

	/* Perform finalisation. */
	database_fsync_data(transaction->database);

	/* Finalisation done.*/
	transaction_to_commited(transaction);
	transaction_log_flush(transaction->database->transaction_log);

	transaction_log_unlock_item(transaction->database->transaction_log, transaction->id);
	/* UNLOCK{ transaction_log transaction } */

	transaction_destroy(transaction);

	return 0;
}

API_PUBLIC int transaction_rollback(transaction_t *transaction) {
	if (transaction->phase != ABORTED)
		transaction_to_failed(transaction);

	transaction_to_aborted(transaction);
	transaction_log_flush(transaction->database->transaction_log);

	transaction_log_unlock_item(transaction->database->transaction_log, transaction->id);
	/* UNLOCK{ transaction_log transaction } */

	transaction_destroy(transaction);

	return 0;
}
